package com.rm.vip.mq.demo.p2p;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;

import javax.jms.*;

/**
 * @ProjectName: mq-demo
 * @Auther: GERRY
 * @Date: 2019/2/21 21:58
 * @Description:
 */
public class QueueConsumer {
    @Test
    public void testQueueConsumer() throws Exception {
        // 创建一个ConnectionFactory对象连接MQ服务器
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)?randomize=true");
        // 创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        // 开启连接
        connection.start();
        // 使用Connection对象创建一个Session对象
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        // 创建一个Destination对象。queue对象
        Queue queue = session.createQueue("test-queue ");
        // 使用Session对象创建一个消费者对象。
        MessageConsumer consumer = session.createConsumer(queue);

        /*for (int i = 1; i <=5; i++) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            System.out.println(textMessage.getText());
            if (i ==3) {
                // 3条之前消费的消息都将被签收
                textMessage.acknowledge();
            }
        }*/

        // 异步接收消息
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                // 打印结果
                TextMessage textMessage = (TextMessage) message;

                /*try {
                    // 需要消费端手动签收消息
                    textMessage.acknowledge();
                } catch (JMSException e) {
                    e.printStackTrace();
                }*/

                String text;
                try {
                    text = textMessage.getText();
                    System.out.println("这是接收到的消息： " + text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        // 等待接收消息
        Thread.currentThread().join(2000);
        session.recover();
        // 关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}
